package com.ads;

import com.alibaba.fastjson.JSONObject;
import com.utils.KafkaUtils;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

/**
 * 统计每个用户的购买频次、购买金额和客单价，并将结果写入到 ClickHouse 中（10分）
 */
public class nine3 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        //从kafka主题读取order_info和order_detail的宽表使用广播关联sku_info的大表
        DataStreamSource<String> map = env.addSource(KafkaUtils.createConsumer("ads_wide_db", "iKun"));

        SingleOutputStreamOperator<JSONObject> stream = map.map(x -> {
            return JSONObject.parseObject(x);
        });
        //stream.print();
        //统计每种商品的销售额、销售量和客单价，并将结果写入到 ClickHouse 中（10分）
        SingleOutputStreamOperator<String> processStream = stream.keyBy(x -> {
            return x.getJSONObject("info").getJSONObject("after").getString("user_id");
        }).process(new KeyedProcessFunction<String, JSONObject, String>() {
            ValueState<Double> moneyState;
            ValueState<Double> numState;

            @Override
            public void open(Configuration parameters) throws Exception {
                //监控状态
                ValueStateDescriptor<Double> money = new ValueStateDescriptor<>("price", Double.class);
                ValueStateDescriptor<Double> num = new ValueStateDescriptor<>("num", Double.class);
                moneyState = getRuntimeContext().getState(money);
                numState = getRuntimeContext().getState(money);
            }

            @Override
            public void processElement(JSONObject jsonObject, KeyedProcessFunction<String, JSONObject, String>.Context context, Collector<String> collector) throws Exception {
                //统计每种商品的销售额、销售量和客单价
                Double money = moneyState.value();
                if (money == null) {
                    money = 0.0;
                    moneyState.update(money);
                }
                Double num = numState.value();
                if (num == null) {
                    num = 0.0;
                    numState.update(num);
                }
                //销售额=销售单价x销售量
                Double price = jsonObject.getJSONObject("detail").getJSONObject("after").getDouble("order_price");
                Double count = jsonObject.getJSONObject("detail").getJSONObject("after").getDouble("sku_num");
                money=money+(price*count);
                moneyState.update(money);
                //销售量
                num=num+ count;
                numState.update(num);
                //客单价
                Double singlePrice = money / num;
                collector.collect(context.getCurrentKey() + "-" + "销售额:" + money + "," + "销售量:" + num + "," + "客单价:" + singlePrice);
            }
        });
        processStream.print();
        env.execute();
    }
}
